package com.hissy.common.activemq;

import java.util.ArrayList;
import java.util.List;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import com.alibaba.fastjson.JSON;

import org.apache.activemq.command.ActiveMQMessage;

/**
 * Created by zhangyanfei on 2017/12/5.
 *
 * @author zhangyanfei
 * @date 2017/12/05
 */
public class QueueService {
    private ActiveMqClientFactory activeMqClientFactory = ActiveMqClientFactory.getInstance();

    public <T> void enqueue(String queueName, T msg) throws JMSException {
        //1.获取连接
        Connection connection = activeMqClientFactory.getActiveMqConnection();
        connection.start();
        //2.获取会话 第一个参数 事物标识  第二个参数签收模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //3.创建一个队列(Destination)
        Queue queue = session.createQueue(queueName);
        //4.创建一个消息发送器
        MessageProducer messageProducer = session.createProducer(queue);
        //5.发送消息
        Message message = new ActiveMQMessage();
        message.setStringProperty("msg", JSON.toJSONString(msg));
        messageProducer.send(message);
        connection.close();
    }

    public Message popMessage(String queueName) throws JMSException {
        //1.获取连接
        Connection connection = activeMqClientFactory.getActiveMqConnection();
        connection.start();
        //2.获取会话 第一个参数 事物标识  第二个参数签收模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //3.创建一个队列(Destination)
        Queue queue = session.createQueue(queueName);
        //4.创建一个消息接收器
        MessageConsumer messageConsumer = session.createConsumer(queue);
        Message message = messageConsumer.receive();
        connection.close();
        return message;
    }

    /**
     * 持续监听并处理队列消息
     * @param queueName 消息队列名称
     * @param clazz 消息体类型
     * @param threadCount 同时处理消息的线程数
     * @param consumer 消息处理程序
     * @param <T>
     */
    public <T> void listen(String queueName, int threadCount, Class<T> clazz,ActionWithException2<T> consumer) {
        listen(queueName, threadCount, body -> {
            T msg = JSON.parseObject(body, clazz);
            consumer.invoke(msg);
        });
    }

    /**
     * 持续监听并处理队列消息
     * @param queueName
     * @param threadCount
     * @param consumer
     */
    public void listen(String queueName, int threadCount, ActionWithException2<String> consumer) {
        Runnable listenAction = () -> {
            while (true) {
                try {
                    exec(queueName, session -> {
                        while (true) {
                            //1.创建一个队列
                            Queue queue = session.createQueue(queueName);
                            //2.创建一个消息接收器
                            MessageConsumer messageConsumer = session.createConsumer(queue);
                            Message queueMsg = messageConsumer.receive();
                            if (queueMsg != null) {
                                String msg = queueMsg.getStringProperty("msg");
                                try {
                                    consumer.invoke(msg);
                                } catch (Exception e) {
                                    continue;
                                }
                            }
                        }
                    });
                } catch (Exception e) {
                    //10秒后重新连接
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException ex2) {
                        break;
                    }
                }
            }
        };

        // 多线程监听
        List<Thread> threadList = new ArrayList<>(threadCount);
        for(int i=0;i<threadCount;++i){
            Thread thread=new Thread(listenAction);
            thread.start();
            threadList.add(thread);
        }
    }

    /**
     * 重试拿到 会话
     * @param queueName
     * @param action
     */
    private void exec(String queueName, ActionWithException2<Session> action) {
        Connection connection = null;
        try {
            connection = activeMqClientFactory.getActiveMqConnection();
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            while (true) {
                try {
                    action.invoke(session);
                    break;
                } catch (Exception e) {
                    //todo 重新获取会话再试
                    throw e;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
